#
# #!/usr/bin/env python
# import threading, time
#
# from kafka import KafkaAdminClient, KafkaConsumer, KafkaProducer
# from kafka.admin import NewTopic
#
#
# class Producer(threading.Thread):
#     def __init__(self):
#         threading.Thread.__init__(self)
#         self.stop_event = threading.Event()
#
#     def stop(self):
#         self.stop_event.set()
#
#     def run(self):
#         producer = KafkaProducer(bootstrap_servers='localhost:9092')
#
#         while not self.stop_event.is_set():
#             print("+++++++")
#             producer.send('my-topic', b"test")
#             producer.send('my-topic', b"\xc2Hola, mundo!")
#             time.sleep(1)
#
#         producer.close()
#
#
# class Consumer(threading.Thread):
#     def __init__(self):
#         threading.Thread.__init__(self)
#         self.stop_event = threading.Event()
#
#     def stop(self):
#         self.stop_event.set()
#
#     def run(self):
#         consumer = KafkaConsumer(bootstrap_servers='localhost:9092',
#                                  auto_offset_reset='earliest',
#                                  consumer_timeout_ms=1000)
#         consumer.subscribe(['my-topic'])
#
#         while not self.stop_event.is_set():
#             for message in consumer:
#                 print(message)
#                 if self.stop_event.is_set():
#                     break
#
#         consumer.close()
#
#
# def main():
#     # Create 'my-topic' Kafka topic
#     try:
#         admin = KafkaAdminClient(bootstrap_servers='localhost:9092')
#
#         topic = NewTopic(name='my-topic',
#                          num_partitions=1,
#                          replication_factor=1)
#         admin.create_topics([topic])
#     except Exception:
#         pass
#
#     tasks = [
#         Producer(),
#         Consumer()
#     ]
#
#     # Start threads of a publisher/producer and a subscriber/consumer to 'my-topic' Kafka topic
#     for t in tasks:
#         t.start()
#
#     time.sleep(100)
#
#     # Stop threads
#     for task in tasks:
#         task.stop()
#
#     for task in tasks:
#         task.join()
#
#
# if __name__ == "__main__":
#     main()


from kafka import KafkaProducer
import json
from test.data import get_registered_user_key
from test.data import get_registered_user_value
import time


def json_serializer(data):
    return json.dumps(data).encode("utf-8")




producer = KafkaProducer(bootstrap_servers = ['192.168.1.160:9092'],
                         value_serializer = json_serializer,
                         key_serializer = json_serializer)


if __name__ == "__main__":
    while True:


        registered_user_key = get_registered_user_key()
        registered_user_value = get_registered_user_value()
        print(registered_user_key)
        producer.send("registered_user",registered_user_value,registered_user_key)
        time.sleep(3)



